package com.github.binarylei.drpc;

import com.github.binarylei.wordcount.bolt.WorldCountBolt;
import com.github.binarylei.wordcount.bolt.WorldReportBolt;
import com.github.binarylei.wordcount.bolt.WorldSplitBolt;
import com.github.binarylei.wordcount.spout.WorldSpout;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.LocalDRPC;
import org.apache.storm.StormSubmitter;
import org.apache.storm.drpc.LinearDRPCTopologyBuilder;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
 * @author: leigang
 * @version: 2018-04-07
 */
public class BasicDRPCTopology {

    public static class ExclamBolt extends BaseBasicBolt {

        @Override
        public void execute(Tuple input, BasicOutputCollector collector) {
            String str = input.getString(1);
            collector.emit(new Values(input.getValue(0), str + "!"));
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("id", "result"));
        }
    }

    public static void main(String[] args) throws InterruptedException, InvalidTopologyException, AuthorizationException, AlreadyAliveException {

        LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("exclamation");
        builder.addBolt(new ExclamBolt(), 3);

        Config conf = new Config();
        if (args == null || args.length ==0) {
            LocalDRPC drpc = new LocalDRPC();
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("drpc-demo", conf, builder.createLocalTopology(drpc));

            for (String word : new String[]{"hello", "goodbye"}) {
                System.out.println(drpc.execute("exclamation", word));
            }
            cluster.shutdown();
            drpc.shutdown();
        } else {
            conf.setNumWorkers(3);
            StormSubmitter.submitTopology(args[0], conf, builder.createRemoteTopology());
        }
    }
}
